package com.ideal.manage.esms.service;

import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.google.common.collect.Lists;
import com.ideal.manage.esms.service.monitor.ESValueSourceAggregation;
import com.ideal.manage.esms.service.monitor.Utils;
import org.elasticsearch.action.search.SearchRequestBuilder;
import org.elasticsearch.action.search.SearchResponse;
import org.elasticsearch.client.transport.TransportClient;
import org.elasticsearch.common.collect.Tuple;
import org.elasticsearch.index.query.BoolQueryBuilder;
import org.elasticsearch.index.query.QueryBuilders;
import org.elasticsearch.search.SearchHit;
import org.elasticsearch.search.aggregations.Aggregation;
import org.elasticsearch.search.aggregations.AggregationBuilder;
import org.elasticsearch.search.aggregations.AggregationBuilders;
import org.elasticsearch.search.aggregations.bucket.MultiBucketsAggregation;
import org.elasticsearch.search.aggregations.bucket.histogram.DateHistogramAggregationBuilder;
import org.elasticsearch.search.aggregations.bucket.histogram.DateHistogramInterval;
import org.elasticsearch.search.aggregations.bucket.histogram.InternalDateHistogram;
import org.elasticsearch.search.aggregations.metrics.InternalNumericMetricsAggregation;
import org.elasticsearch.search.aggregations.metrics.avg.InternalAvg;
import org.elasticsearch.search.aggregations.support.ValuesSourceAggregationBuilder;
import org.elasticsearch.search.sort.SortBuilders;
import org.elasticsearch.search.sort.SortOrder;
import org.joda.time.DateTime;
import org.joda.time.DateTimeZone;
import org.joda.time.format.DateTimeFormat;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import java.math.BigDecimal;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

@Service
public class ElasticsearchService {
    private Logger LOG = LoggerFactory.getLogger(ElasticsearchService.class);

    @Autowired
    private TransportClient transportClient;

    protected String buildKey(Map<String, Object> map) {
        return String.format("%s-%s-%s-%s", map.get("id").toString().split("_")[0], map.get("ip"), map.get("port"), map.get("macid"));
    }


    public JSONArray searchNewest(BoolQueryBuilder boolQueryBuilder, int size) {
        SearchRequestBuilder prepareSearch = transportClient.prepareSearch();
        prepareSearch.setQuery(boolQueryBuilder
                .filter(QueryBuilders.existsQuery("time"))
                .filter(QueryBuilders.existsQuery("port"))
                .filter(QueryBuilders.existsQuery("macid"))
                .filter(QueryBuilders.existsQuery("ip")));
        prepareSearch.setSize(size);
        prepareSearch.addSort("time", SortOrder.DESC);
        SearchResponse searchResponse = prepareSearch.execute().actionGet();
        if (searchResponse.getHits().getHits().length == 0) {
            return null;
        }
        JSONArray list = new JSONArray();
        for (SearchHit hit : searchResponse.getHits().getHits()) {
            Map<String, Object> hitSourceAsMap = hit.getSourceAsMap();
            list.add(new JSONObject(hitSourceAsMap));
        }
        return list;
    }

    public JSONObject searchNewest(BoolQueryBuilder boolQueryBuilder, MatchFunction func) {
        SearchRequestBuilder prepareSearch = transportClient.prepareSearch();
        prepareSearch.setQuery(boolQueryBuilder
                .filter(QueryBuilders.existsQuery("time"))
                .filter(QueryBuilders.existsQuery("port"))
                .filter(QueryBuilders.existsQuery("macid"))
                .filter(QueryBuilders.existsQuery("ip")));
        prepareSearch.setSize(1);
        prepareSearch.addSort("time", SortOrder.DESC);
        SearchResponse searchResponse = prepareSearch.execute().actionGet();
        if (searchResponse.getHits().getHits().length == 0) {
            return null;
        }
        JSONObject r = new JSONObject();
        for (SearchHit hit : searchResponse.getHits().getHits()) {
            Map<String, Object> hitSourceAsMap = hit.getSourceAsMap();
            for (String key : hitSourceAsMap.keySet()) {
                if (func.isMatch(key)) {
                    r.put(key, hitSourceAsMap.get(key));
                }
            }
        }
        return r;
    }

    public JSONObject searchByMatchFunctionNews(BoolQueryBuilder boolQueryBuilder, MatchFunction func) {
        List<JSONObject> list = this.searchByMatchFunction(boolQueryBuilder, func);
        if (list.isEmpty()) {
            return null;
        }
        return list.get(0);
    }

    public void aggregationAvg(BoolQueryBuilder boolQueryBuilder,
                               DateHistogramInterval interval,
                               AggregationResultFunction resultFunction,
                               String... filedNames) {
        this.aggregation(boolQueryBuilder, interval, new ESValueSourceAggregation("avg"), resultFunction, filedNames);
    }

    public void aggregationSum(BoolQueryBuilder boolQueryBuilder,
                               DateHistogramInterval interval,
                               AggregationResultFunction resultFunction,
                               String... filedNames) {
        this.aggregation(boolQueryBuilder, interval, new ESValueSourceAggregation("sum"), resultFunction, filedNames);
    }

    public void aggregation(BoolQueryBuilder boolQueryBuilder,
                            DateHistogramInterval interval,
                            ESValueSourceAggregation funcAggregation, AggregationResultFunction resultFunction,
                            String... filedNames) {
        SearchRequestBuilder prepareSearch = transportClient.prepareSearch();
        prepareSearch.setQuery(boolQueryBuilder
                .filter(QueryBuilders.existsQuery("time"))
                .filter(QueryBuilders.existsQuery("port"))
                .filter(QueryBuilders.existsQuery("macid"))
                .filter(QueryBuilders.existsQuery("ip")));
        prepareSearch.setSize(0);
        prepareSearch.addSort("time", SortOrder.DESC);

        DateHistogramAggregationBuilder aggregationBuilder = AggregationBuilders.dateHistogram("time_agg")
                .dateHistogramInterval(interval).field("time");

        for (String filedName : filedNames) {
            aggregationBuilder.subAggregation(funcAggregation.func(filedName + "_val").field(filedName));
        }
        prepareSearch.addAggregation(aggregationBuilder);
        SearchResponse searchResponse = prepareSearch.execute().actionGet();
        Map<String, Aggregation> aggregations = searchResponse.getAggregations().asMap();
        if (aggregations.size() == 0) {
            return;
        }
        for (String filedName : filedNames) {
            InternalDateHistogram aggregation = (InternalDateHistogram) aggregations.get("time_agg");
            for (MultiBucketsAggregation.Bucket bucket : aggregation.getBuckets()) {
                InternalNumericMetricsAggregation.SingleValue val = bucket.getAggregations().get(filedName + "_val");
                LOG.info(val + "                   ");
                double avg = Utils.halfUp(val.value());
                DateTime dateTime = (DateTime) bucket.getKey();
                resultFunction.process(filedName, dateTime.toDateTime(DateTimeZone.getDefault()), avg);
            }
        }
    }

    public Map<String, JSONObject> searchByMatchFunctionAsMap(BoolQueryBuilder boolQueryBuilder, MatchFunction func) {
        Map<String, JSONObject> map = new HashMap<>();
        SearchRequestBuilder prepareSearch = transportClient.prepareSearch();
        prepareSearch.setQuery(boolQueryBuilder
                .filter(QueryBuilders.existsQuery("time"))
                .filter(QueryBuilders.existsQuery("port"))
                .filter(QueryBuilders.existsQuery("macid"))
                .filter(QueryBuilders.existsQuery("ip")));
        prepareSearch.setSize(10000);
        SearchResponse searchResponse = prepareSearch.execute().actionGet();
        if (searchResponse.getHits().getHits().length == 0) {
            return map;
        }
        for (SearchHit hit : searchResponse.getHits().getHits()) {
            Map<String, Object> hitSourceAsMap = hit.getSourceAsMap();
            String hitKey = this.buildKey(hitSourceAsMap);
            JSONObject jsonObject = map.get(hitKey);
            if (jsonObject == null) {
                jsonObject = new JSONObject();
                jsonObject.put("time", hitSourceAsMap.get("time"));
                jsonObject.put("port", hitSourceAsMap.get("port"));
                jsonObject.put("macid", hitSourceAsMap.get("macid"));
                jsonObject.put("ip", hitSourceAsMap.get("ip"));
            }
            for (String key : hitSourceAsMap.keySet()) {
                if (func.isMatch(key)) {
                    jsonObject.put(key, hitSourceAsMap.get(key));
                }
            }
            if (jsonObject.keySet().size() > 4) {
                if (map.containsKey(hitKey)) {
                    jsonObject.putAll(map.get(hitKey));
                }
                map.put(hitKey, jsonObject);
            }
        }
        return map;
    }

    public List<JSONObject> searchByMatchFunction(BoolQueryBuilder boolQueryBuilder, MatchFunction func) {
        List<JSONObject> r = new ArrayList<>();
        Map<String, JSONObject> map = this.searchByMatchFunctionAsMap(boolQueryBuilder, func);
        for (Map.Entry<String, JSONObject> entry : map.entrySet()) {
            r.add(entry.getValue());
        }
        if (LOG.isDebugEnabled()) {
            LOG.debug("return {}", r);
        }
        return r;
    }


    public interface AggregationResultFunction {
        void process(String filedName, DateTime time, double val);
    }
}
